package com.hexinfo.dmpro.sparing.async;

import cn.hutool.core.io.file.FileAppender;
import cn.hutool.core.util.StrUtil;
import com.hexinfo.dmpro.common.utils.LogFileUtil;
import com.hexinfo.dmpro.common.utils.ScanCommonConstants;
import com.hexinfo.dmpro.sparing.dto.TableSyncConfDTO;
import com.hexinfo.dmpro.sparing.exception.ExecException;
import com.hexinfo.dmpro.sparing.model.*;
import com.hexinfo.dmpro.sparing.service.HotStandbyLogTableService;
import com.hexinfo.dmpro.sparing.service.HotStandbyTableService;
import com.hexinfo.dmpro.sparing.service.TableSyncMethodService;
import lombok.Cleanup;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Component;

import javax.sql.DataSource;
import java.io.File;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.*;
import java.util.concurrent.Future;

@Component
@Slf4j
@RequiredArgsConstructor(onConstructor_ = {@Lazy, @Autowired})
public class AsyncFailTableRetrySource {

    private final LogFileUtil logFileUtil;
    private final HotStandbyLogTableService hotStandbyLogTableService;
    private final TableSyncMethodService tableSyncMethodService;

    @Autowired
    @Qualifier("connPoolImpalaSource")
    private DataSource connPoolImpalaSource;
    @Autowired
    @Qualifier("connPoolImpalaTarget")
    private DataSource connPoolImpalaTarget;

    /**
     * distcp命令shell脚本
     */
    @Value("${hdfs.heatBackup}")
    private String heatBackup;
    /**
     * 库环境前缀
     */
    @Value("${hdfs.tarPrefix}")
    private String tarPrefix;

    /**
     * 单表热备同步全流程（异步）
     */
    @Async("tableBackupThreadPool")
    public Future<Boolean> tableHotStandby(HotStandbyLogTable hotStandbyLogTable,TableSyncConfDTO tableSyncConfDTO) {
        try {
            //修改表重跑状态,获取日志对象
            upTaskLog(hotStandbyLogTable,tableSyncConfDTO);
            //测试环境还是生产环境
            String dbTbName = isUatOrPrd(hotStandbyLogTable);
            //判断是否是全量热备
            Boolean aboolean = allHotStandby(hotStandbyLogTable,tableSyncConfDTO,dbTbName);
            //修改表结束重跑状态
            upEndTask(hotStandbyLogTable, aboolean);
            return new AsyncResult<>(aboolean);
        }catch (Exception e){
            log.error("【异常】单表热备失败重跑");
            log.error(e.getMessage(),e);
            upTask(hotStandbyLogTable,ScanCommonConstants.ClusterName.CPSB.value);
            return new AsyncResult<>(false);
        }finally {

        }
    }

    /**
     * 单表热备同步全流程（串行）
     */
    public Boolean tableHotStandbySync(HotStandbyLogTable hotStandbyLogTable,TableSyncConfDTO tableSyncConfDTO) {
        try {
            //修改表重跑状态，获取日志对象
            upTaskLog(hotStandbyLogTable,tableSyncConfDTO);
            //测试环境还是生产环境
            String dbTbName = isUatOrPrd(hotStandbyLogTable);
            //全表热备
            Boolean aboolean = allHotStandby(hotStandbyLogTable,tableSyncConfDTO,dbTbName);
            //修改表结束重跑状态
            upEndTask(hotStandbyLogTable, aboolean);
            return aboolean;
        }catch (Exception e){
            log.error("【异常】单表热备失败重跑");
            log.error(e.getMessage(),e);
            upTask(hotStandbyLogTable,ScanCommonConstants.ClusterName.CPSB.value);
            return false;
        }
    }

    /**
     * 全表热备
     */
    public boolean allHotStandby(HotStandbyLogTable hotStandbyLogTable,TableSyncConfDTO tableSyncConfDTO,String dbTbName){
        try{
            logFileUtil.appendLog(hotStandbyLogTable.getHeatLog(),"-------------------------------【数据同步】",false);
            //数据同步
            dataSyncAll(hotStandbyLogTable,tableSyncConfDTO);
            logFileUtil.appendLog(hotStandbyLogTable.getHeatLog(),"-------------------------------【刷目标表】",false);
            //刷目标表
            refreshTable(hotStandbyLogTable,dbTbName);
            logFileUtil.appendLog(hotStandbyLogTable.getHeatLog(),"-------------------------------【数据比对】",false);
            //数据比对
            dataCompare(hotStandbyLogTable,dbTbName);
            return true;
        }catch (Exception e){
            return false;
        }
    }

    //==============================================数据同步=============================================================

    /**
     * 数据同步（全表）
     */
    public void dataSyncAll(HotStandbyLogTable hotStandbyLogTable,TableSyncConfDTO tableSyncConfDTO){
        String jqHdfs = tableSyncConfDTO.getJQHdfsPrefix() + hotStandbyLogTable.getDbName()+".db/"+hotStandbyLogTable.getTbName();
        String wgHdfs = tableSyncConfDTO.getWGHdfsPrefix() + hotStandbyLogTable.getUatDbName()+".db/"+hotStandbyLogTable.getTbName();
        //完整的distcp的sh脚本命令
        String distcpShell = heatBackup +" "+ jqHdfs +" "+ wgHdfs;
        int code = tableSyncMethodService.execDistcpCode(distcpShell, hotStandbyLogTable.getHeatLog());
    }

    //==============================================刷目标表=============================================================

    /**
     * 刷目标表
     */
    public void refreshTable(HotStandbyLogTable hotStandbyLogTable,String dbTbName){
        //① impala刷新元数据
        String alterSql = "alter table {tableName} recover partitions";
        String sqlAlter = alterSql.replace("{tableName}", dbTbName);
        tableSyncMethodService.refreshDataImpala(sqlAlter,hotStandbyLogTable.getHeatLog());
        //② 刷新impala目标表
        String sql = "invalidate metadata {dbTbName}";
        String refreshSql = sql.replace("{dbTbName}", dbTbName);
        tableSyncMethodService.execSqlImpala(refreshSql,hotStandbyLogTable.getHeatLog());
    }

    //==============================================数据比对=============================================================

    private String selectWYSql = "select count(1) as count from {}";
    /**
     * 数据比对
     */
    public void dataCompare(HotStandbyLogTable hotStandbyLogTable,String dbTbName){
        try{
            String sql = StrUtil.format(selectWYSql,hotStandbyLogTable.getDbTbName());
            String tarSql = StrUtil.format(selectWYSql,dbTbName);
            @Cleanup
            Connection souImpalaConn = connPoolImpalaSource.getConnection();
            Set<String> souSet = selectDateTable(souImpalaConn, sql, "count", hotStandbyLogTable.getHeatLog());
            @Cleanup
            Connection tarImpalaConn = connPoolImpalaTarget.getConnection();
            Set<String> tarSet = selectDateTable(tarImpalaConn, tarSql, "count", hotStandbyLogTable.getHeatLog());
            String souCount = new ArrayList<>(souSet).get(0);
            String tarCount = new ArrayList<>(tarSet).get(0);
            long souCountTotal = Long.parseLong(souCount);
            long tarCountTotal = Long.parseLong(tarCount);
            //修改表的count总条数
            hotStandbyLogTable.setSourceCountNumber(souCountTotal);
            hotStandbyLogTable.setTargetCountNumber(tarCountTotal);
            if (souCountTotal == tarCountTotal){
                logFileUtil.appendLog(hotStandbyLogTable.getHeatLog(),"【比对成功】count总大小："+souCountTotal,false);
            }else {
                logFileUtil.appendLog(hotStandbyLogTable.getHeatLog(),"【比对失败】源count总大小："+souCountTotal+"，目标count总大小："+tarCountTotal,false);
                throw new ExecException("10005");
            }
        }catch (Exception e){
            logFileUtil.appendLog(hotStandbyLogTable.getHeatLog(),"【异常】获取count总数异常："+e.toString(),false);
            throw new ExecException("10005");
        }
    }

    /**
     * 查询每张表的日期分区、查询count总量
     */
    private Set<String> selectDateTable(Connection conn,String sql,String dateColName,FileAppender heatLog) throws Exception {
        Set<String> dateValues = new HashSet<>();
        try{
            @Cleanup
            Statement souSt = conn.createStatement();
            @Cleanup
            ResultSet resultSet = souSt.executeQuery(sql);
            //获取日期分区
            while (resultSet.next()) {
                dateValues.add(resultSet.getString(dateColName));
            }
            logFileUtil.appendLog(heatLog,sql,true);
        }catch (Exception e){
            log.error("【异常】"+sql);
            log.error(e.getMessage(),e);
            logFileUtil.appendLog(heatLog,"【异常】"+sql,true);
            logFileUtil.appendLog(heatLog,"【异常】"+e.toString(),false);
            throw new Exception(e.getMessage());
        }
        return dateValues;
    }

    //=================================================================================================================

    /**
     * 测试环境还是生产环境
     */
    public String isUatOrPrd(HotStandbyLogTable hotStandbyLogTable){
        String[] split1 = hotStandbyLogTable.getDbTbName().split("\\.");
        String dbName = split1[0];
        String tbName = split1[1];
        String uatDbName = dbName;
        if (StrUtil.isNotBlank(tarPrefix)){
            uatDbName =  tarPrefix + dbName;
        }
        hotStandbyLogTable.setUatDbName(uatDbName);
        hotStandbyLogTable.setDbName(dbName);
        hotStandbyLogTable.setTbName(tbName);
        return uatDbName +"."+ tbName;
    }

    /**
     * 修改表重跑状态,获取日志对象
     */
    public void upTaskLog(HotStandbyLogTable hotStandbyLogTable,TableSyncConfDTO tableSyncConfDTO){
        upTask(hotStandbyLogTable,ScanCommonConstants.ClusterName.CPZ.value);
        FileAppender heatLog = new FileAppender(new File(hotStandbyLogTable.getLogPath()), 100, true);
        hotStandbyLogTable.setHeatLog(heatLog);
        log.warn("-----------------------失败重跑ING-----------------------------");
        LogFileUtil.appendLog(hotStandbyLogTable.getHeatLog(),"-----------------------------------------------------------",false);
        logFileUtil.appendLog(hotStandbyLogTable.getHeatLog(),"-----------------------------------------------------------",true);
        logFileUtil.appendLog(hotStandbyLogTable.getHeatLog(),"-----------------------失败重跑-----------------------------",false);
    }

    /**
     * 修改表结束重跑状态
     */
    public void upEndTask(HotStandbyLogTable hotStandbyLogTable, Boolean aboolean){
        if (aboolean){
            hotStandbyLogTable.setTaskState(ScanCommonConstants.ClusterName.CG.value);
            upTask(hotStandbyLogTable,ScanCommonConstants.ClusterName.CPCG.value);
        }else {
            upTask(hotStandbyLogTable,ScanCommonConstants.ClusterName.CPSB.value);
        }
    }

    /**
     * 修改重跑状态
     */
    public void upTask(HotStandbyLogTable hotStandbyLogTable,String state){
        hotStandbyLogTable.setRerunState(state);
        hotStandbyLogTableService.updateById(hotStandbyLogTable);
    }

}